package juc.executors;

import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class CompletableFutureExample1 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //test1();
        //test2();
        //test3();
        test4();
    }


    /**
     * CompletableFuture是Future和ExecutorService的结合，解决了此种阻塞问题
     */
    private static void test1() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        //异步调用
        Future<?> future = executorService.submit(() -> {
            timeSleep(1);
        });

        //阻塞
        future.get();
        //相当于block
        while (future.isDone()) {
        }
        System.out.println("done");
    }

    private static void test2() throws ExecutionException, InterruptedException {
        /**
         * CompletableFuture是Future和ExecutorService的结合，有一个内置的ExecutorService
         * CompletableFuture代表的是一个Future，没有关于ExecutorService的一些方法，不能管理ExecutorService
         * 因此在CompletableFuture里的ExecutorService都是守护线程，一般不写在main方法中
         *      ForkJoinPool#registerWorker(ForkJoinWorkerThread wt){wt.setDaemon(true)}
         */
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {
            timeSleep(3);
        }).whenComplete((v, t) -> {
            System.out.println("done");
        });

        System.out.println("i am not blocked");
        //使用join阻塞main线程
        Thread.currentThread().join();
    }

    private static void test3() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        List<Callable<Integer>> tasks = IntStream.range(0, 5).boxed()
                .map(i -> (Callable<Integer>) () -> get()).collect(Collectors.toList());

        executorService.invokeAll(tasks).stream().map(future -> {
            try {
                //阻塞，只有第一阶段的任务执行完成后才会执行第二阶段
                return future.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        }).parallel().forEach(CompletableFutureExample1::display);
    }

    private static int display(int data) {
        int value = ThreadLocalRandom.current().nextInt(7);
        System.out.println(Thread.currentThread().getName() + " display will be sleep " + value);
        timeSleep(value);
        System.out.println(Thread.currentThread().getName() + " display execute done " + data);
        return value;
    }

    private static int get() {
        int value = ThreadLocalRandom.current().nextInt(7);
        System.out.println(Thread.currentThread().getName() + " get will be sleep " + value);
        timeSleep(value);
        System.out.println(Thread.currentThread().getName() + " get execute done " + value);
        return value;
    }

    private static void test4() throws ExecutionException, InterruptedException {
        IntStream.range(0, 5).boxed().forEach(i ->
                CompletableFuture.supplyAsync(CompletableFutureExample1::get)
                        .thenAccept(CompletableFutureExample1::display)
                        .whenComplete((v, t) -> System.out.println(i + "done")));

        //使用join阻塞main线程
        Thread.currentThread().join();
    }


    /**
     * sleep the specify seconds
     */
    private static void timeSleep(long time) {
        try {
            TimeUnit.SECONDS.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
